/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */
package com.eluup.flume.sink.elasticsearch.client;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.eluup.flume.sink.elasticsearch.ElasticSearchEventSerializer;
import com.eluup.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory;
import com.eluup.flume.sink.elasticsearch.ElasticSearchScriptSerializer;
import com.eluup.flume.sink.elasticsearch.ElasticSearchSink;

/**
 * Internal ElasticSearch client factory. Responsible for creating instance of
 * ElasticSearch clients.
 */
public class ElasticSearchClientFactory {
	public static final String TransportClientScript = "script";
	public static final String TransportClientXPack = "x-pack-transport";
	public static final String TransportClientRobust = "robust-transport";
	public static final String TransportClient = "transport";
	public static final String RestClient = "rest";
	private static final Logger logger = LoggerFactory.getLogger(ElasticSearchClientFactory.class);
	/**
	 * @param clientType
	 *            String representation of client type
	 * @param hostNames
	 *            Array of strings that represents hosntames with ports
	 *            (hostname:port)
	 * @param clusterName
	 *            Elasticsearch cluster name used only by Transport Client
	 * @param serializer
	 *            Serializer of flume events to elasticsearch documents
	 * @return
	 */
	public ElasticSearchClient getClient(String clientType, String[] hostNames, String clusterName,
	        ElasticSearchEventSerializer serializer,
	        ElasticSearchIndexRequestBuilderFactory indexBuilder) throws NoSuchClientTypeException {
			
		
		if (clientType.equalsIgnoreCase(TransportClientScript) && serializer != null) {
			if (!(serializer instanceof ElasticSearchScriptSerializer)) {
				logger.error("Serialized object type error! Please use [com.eluup.flume.sink.elasticsearch.ElasticSearchScriptSerializer]!");
				throw new NoSuchClientTypeException();
			}
			return new ElasticSearchTransportScriptClient(hostNames, clusterName, (ElasticSearchScriptSerializer) serializer);
		} else if (clientType.equalsIgnoreCase(TransportClientScript) && indexBuilder != null) {
			return new ElasticSearchTransportScriptClient(hostNames, clusterName);
		} else if (clientType.equalsIgnoreCase(TransportClientRobust) && serializer != null) {
			return new ElasticSearchTransportRobustClient(hostNames, clusterName, serializer);
		} else if (clientType.equalsIgnoreCase(TransportClientRobust) && indexBuilder != null) {
			return new ElasticSearchTransportRobustClient(hostNames, clusterName, indexBuilder);
			
		} else if (clientType.equalsIgnoreCase(TransportClientXPack) && serializer != null) {
			return new ElasticSearchTransportXPackClient(hostNames, clusterName, serializer);
		} else if (clientType.equalsIgnoreCase(TransportClientXPack) && indexBuilder != null) {
			return new ElasticSearchTransportXPackClient(hostNames, clusterName, indexBuilder);

		} else if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
			return new ElasticSearchTransportClient(hostNames, clusterName, serializer);
		} else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) {
			return new ElasticSearchTransportClient(hostNames, clusterName, indexBuilder);
			
		} else if (clientType.equalsIgnoreCase(RestClient) && serializer != null) {
			return new ElasticSearchRestClient(hostNames, serializer);
		}
		throw new NoSuchClientTypeException();
	}

	/**
	 * Used for tests only. Creates local elasticsearch instance client.
	 *
	 * @param clientType
	 *            Name of client to use
	 * @param serializer
	 *            Serializer for the event
	 * @param indexBuilder
	 *            Index builder factory
	 * @return Local elastic search instance client
	 */
	public ElasticSearchClient getLocalClient(String clientType,
	        ElasticSearchEventSerializer serializer,
	        ElasticSearchIndexRequestBuilderFactory indexBuilder) throws NoSuchClientTypeException {
		if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
			return new ElasticSearchTransportClient(serializer);
		} else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) {
			return new ElasticSearchTransportClient(indexBuilder);
		} else if (clientType.equalsIgnoreCase(RestClient)) {
		}
		throw new NoSuchClientTypeException();
	}
}
